package com.exchange.controller.app.trade;

import java.math.BigDecimal;
import java.text.ParseException;
import java.util.Date;
import java.util.List;
import java.util.Map;
import java.util.function.Function;
import java.util.stream.Collectors;
import com.exchange.common.model.MarketConfig;
import com.exchange.common.model.User;
import com.exchange.common.model.UserWallet;
import com.exchange.common.model.dto.CancelOrderParam;
import com.exchange.common.model.MatchOrder;
import com.exchange.common.model.dto.Order;
import com.exchange.interceptor.EnumError;
import com.exchange.plugin.kafka.Kafka;
import com.exchange.plugin.threadpool.ThreadPoolKit;
import com.exchange.util.BizUtils;
import com.exchange.util.SnowflakeIdWorker;
import com.jfinal.kit.Ret;
import com.jfinal.plugin.activerecord.Db;
import com.jfinal.plugin.activerecord.Page;
import com.jfinal.plugin.activerecord.Record;

import cn.hutool.core.bean.BeanUtil;
import cn.hutool.core.util.NumberUtil;
import cn.hutool.core.util.StrUtil;

public class TradeService {
	
	int pageSize = 20;
	
	public Ret business(User user, String coinTeam,Boolean isBuy,BigDecimal price,BigDecimal num) throws ParseException {
    	
		List<MarketConfig> coinTeams = MarketConfig.dao.findByCache("coinTeams", "coinTeams", "SELECT * FROM t_market_config WHERE state = 1 ");
    	Map<String, MarketConfig> coinTeamMap = coinTeams.stream().collect(Collectors.toMap(MarketConfig::getCoinTeam, Function.identity() ));
    	//交易对是否存在
    	if(!coinTeamMap.containsKey(coinTeam)) {
    		return Ret.fail("msg", EnumError.COIN_TEAM_NULL.getCode());
    	}
        // 验证是否开盘
    	MarketConfig market = coinTeamMap.get(coinTeam);
        if (!BizUtils.isOpen(market)) {
        	return Ret.fail("msg", EnumError.NOT_OPEN_QUOTA.getCode());
        }
        // 是否c1实名认证
        if(user.getVerified() < 3) {
        	return Ret.fail("msg", EnumError.C1_V_NO_CONDUCT.getCode());
        }
        // 通过买卖判定是哪种钱包
        String coin = isBuy ? market.getMarket() : market.getCoin();
        // 验证余额是否足够
        // 先计算手续费
        BigDecimal total = NumberUtil.mul(num, price);// 总额
        BigDecimal znum = isBuy ? total : num;
        BigDecimal fee = NumberUtil.mul(znum,market.getRate());//手续费(出去什么币种收什么币种)
        Date date = new Date();
        //手续费币种
        if(StrUtil.isBlank(market.getRateCoin())) {
        	// 通过coin和userid获取用户钱包
            UserWallet userWallet = UserWallet.dao.findFirst("SELECT * FROM t_user_wallet WHERE userId = ? AND coin = ? ", user.getId(), coin);
            if(!NumberUtil.isGreaterOrEqual(userWallet.getBalance(), NumberUtil.add(znum,fee))) {
            	return Ret.fail("msg", EnumError.WALLET_INSUFFICIENT_FUNDS.getCode());
            }
            //冻结
        	userWallet.setBalance(NumberUtil.sub(userWallet.getBalance(), NumberUtil.add(znum,fee)))
        	.setFrozenBalance(NumberUtil.add(userWallet.getFrozenBalance(), NumberUtil.add(znum,fee)))
        	.setUpdateTime(date).save();
            
        }else {
        	UserWallet userWallet = UserWallet.dao.findFirst("SELECT * FROM t_user_wallet WHERE userId = ? AND coin = ? ", user.getId(), coin);
            if(!NumberUtil.isGreaterOrEqual(userWallet.getBalance(), znum)) {
            	return Ret.fail("msg", EnumError.WALLET_INSUFFICIENT_FUNDS.getCode());
            }
            UserWallet rateWallet = UserWallet.dao.findFirst("SELECT * FROM t_user_wallet WHERE userId = ? AND coin = ? ", user.getId(), market.getRateCoin());
            if(!NumberUtil.isGreaterOrEqual(rateWallet.getBalance(), fee)) {
            	return Ret.fail("msg", EnumError.WALLET_INSUFFICIENT_FUNDS.getCode());
            }
            //冻结
            userWallet.setBalance(NumberUtil.sub(userWallet.getBalance(), znum))
        	.setFrozenBalance(NumberUtil.add(userWallet.getFrozenBalance(), znum))
        	.setUpdateTime(date).save();
            rateWallet.setBalance(NumberUtil.sub(rateWallet.getBalance(), fee))
        	.setFrozenBalance(NumberUtil.add(rateWallet.getFrozenBalance(), fee))
        	.setUpdateTime(date).save();
        }
        
        Long orderNum = SnowflakeIdWorker.generateId();
        //异步存库
        MatchOrder mo = new MatchOrder().setAlterTime(date).setCreateTime(date).setCoin(coin).setMarket(market.getCoin())
		.setFee(fee).setFeeSymbol(StrUtil.isBlank(market.getRateCoin())?coin:market.getRateCoin())				
		.setIsBuy(isBuy).setIsMarket(false).setNum(num).setOrderNumber(orderNum.toString())
		.setPrice(price).setState(0).setTotalPrice(total).setUserId(user.getId()).setVersion(1);
		if(isBuy) {
			mo.setSurplusPrice(total);
		}else {
			mo.setSurplusNum(num);
		}
		ThreadPoolKit.submit(new Runnable() {
			public void run() {
				mo.save();
			}
		});
        // TODO 生成订单传入kafka
        Order kafkaOrder = new Order();
        kafkaOrder.setId(orderNum);
        kafkaOrder.setUid(user.getId().longValue());
        kafkaOrder.setIsBuy(isBuy);
        kafkaOrder.setPrice(price);
        kafkaOrder.setNumber(num);
        kafkaOrder.setIsMarket(Boolean.FALSE);//true:市价；false：限价
        kafkaOrder.setTotalPrice(BigDecimal.ONE);
        if (!kafkaOrder.getIsMarket()) {
			kafkaOrder.setTotalPrice(total);
		}
        kafkaOrder.setCoinTeam(coinTeam);
        kafkaOrder.setDecimalNumber(8);
        kafkaOrder.setFinishNumber(BigDecimal.ZERO);
        kafkaOrder.setUnFinishNumber(num);
        kafkaOrder.setState(0);
        kafkaOrder.setCreateTime(date);
        if (kafkaOrder.getIsBuy()) {
        	kafkaOrder.setSurplusFrozen(total);// 剩余冻结
		} else {
			kafkaOrder.setSurplusFrozen(num);// 剩余冻结
		}
        //kafka消息队列
        Kafka.send(Kafka.matchOrder, Kafka.matchOrder, kafkaOrder.getId(), kafkaOrder);
        
		return Ret.ok("trderOrder",mo);
    }
	
	public Ret revoke(User user,String orderNum) {
		//查询用户订单
		Record r = Db.findFirst("SELECT orderNumber,coin,market FROM t_match_order WHERE orderNumber = ? AND userId = ? AND state IN(0,1)",orderNum,user.getId());
        if(BeanUtil.isEmpty(r)) {
        	return Ret.fail("msg", EnumError.TRADE_REVOKE_FAIL.getCode());
        }
		//kafka消息队列
		CancelOrderParam c = new CancelOrderParam();
		c.setId(Long.parseLong(r.getStr("orderNumber")));
		c.setIsBuy(true);
		c.setCoinTeam(r.getStr("coin")+r.getStr("market"));
        Kafka.send(Kafka.revokeOrder, Kafka.matchOrder, c.getId(), c);
        
        return Ret.ok();
    }
	
	public Ret activesPage(User user, String coin, String market, Boolean isBuy, Date from, Date to, Integer pageNumber){
		StringBuffer sqlPara = new StringBuffer("FROM t_match_order WHERE userId = ? ");
		
		if(StrUtil.isNotBlank(coin) && StrUtil.isNotBlank(market)) {
			sqlPara.append("AND coin = ").append(coin).append("AND market = ").append(market);
		}
		if(isBuy != null) {
			sqlPara.append("AND isBuy = ").append(isBuy);
		}
		if(from != null && to != null) {
			sqlPara.append("AND createTime > ").append(from).append(" AND ").append(to).append(" > createTime ");
		}
		Page<MatchOrder> activesPage = MatchOrder.dao.paginate(pageNumber, pageSize, "select * ", sqlPara.toString(),user.getId());
		return Ret.ok("activesPage",activesPage);
    }
	
	
	
	
	
}
